package com.geek.gkpush.framework.mq;

import com.geek.gkpush.util.json.JacksonUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;

import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.ArrayList;
import java.util.List;

/**
 * <h1>自定义类，实现 {@linkplain MessageListenerConcurrently} rocketmq 的源码类，主要添加一个 mqName 的方法实现、捕获返回重试机制，易于开发理解。</h1><br/>
 *
 * <h1><br/><a href="https://u.geekbang.org/subject/arch3/100614201">Java 业务架构实战营：https://u.geekbang.org/subject/arch3/100614201</a></h1><br/><h1><a href="https://gitee.com/ylimhhmily/BizArchPracticeCamp">Java 业务架构实战营代码地址：https://gitee.com/ylimhhmily/BizArchPracticeCamp</a></h1><br/><h1><a href="https://u.geekbang.org/subject/java4th/1001148?source=app_share">高级 Java 工程体系课 2023 版：https://u.geekbang.org/subject/java4th/1001148?source=app_share</a></h1><br/><h1><a href="https://time.geekbang.org/column/intro/100312101?tab=catalog">Dubbo 源码剖析与实战：https://time.geekbang.org/column/intro/100312101?tab=catalog</a></h1>
 *
 * @author hmilyylimh
 * ^_^
 * @version 0.0.1
 * ^_^
 * @date 2023/10/2 09:24
 */
@Slf4j
public abstract class AbstractMessageListenerConcurrently<T> implements MessageListenerConcurrently {

    /**
     * <h2>监听的 Q 名称，该方法仅仅只是用于查看日志使用，方便从日志上看出属于哪个 Q 业务而已，没有其他处理业务逻辑的地方。</h2>
     *
     * @throws
     * @param:
     * @return: java.lang.String
     * @author hmilyylimh
     * @date: 2023/9/23
     */
    public abstract String getTopicName();

    /**
     * <h2>只需要关注业务逻辑处理，异常的话就往外抛。</h2>
     *
     * @throws
     * @param: msgs
     * @param: context
     * @return: org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus
     * @author hmilyylimh
     * @date: 2023/10/2
     */
    protected abstract void doConsumeMessage(List<T> msgList);

    /**
     * <h2>当发生异常的时候，需要业务方自行善后处理。</h2>
     *
     * @throws
     * @param: msgList
     * @return: void
     * @author hmilyylimh
     * @date: 2023/10/2
     */
    protected boolean onException(List<T> msgList) {
        return false;
    }

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                    ConsumeConcurrentlyContext context) {
        List<T> msgList = null;
        try {
            // 获取当前泛型类，即 AbstractMqListener<T>
            ParameterizedType parameterizedType = (ParameterizedType) this.getClass().getGenericSuperclass();

            // 当前泛型类只有一个参数T, 此type为Class<T>
            Type msgType = parameterizedType.getActualTypeArguments()[0];

            // 解析数据
            msgList = new ArrayList<>(msgs.size());
            for (MessageExt msg : msgs) {
                byte[] body = msg.getBody();
                // 将 recvMsg 解析为 T 类型的数据，如果为 String 类型的话，就不用转了，直接使用
                if (String.class.equals(msgType)) {
                    msgList.add((T) new String(body));
                } else {
                    msgList.add(JacksonUtils.fromJson(new String(body), msgType));
                }
            }

            // 触发业务逻辑实现体逻辑
            doConsumeMessage(msgList);

            // 没有什么报错的话，那就返回成功
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        } catch (Exception e) {
            log.error("doConsumeMessage error.", e);

            // 如果异常的话，看需要怎么善后处理，如果 onException 处理成功的话，那就直接返回消费成功，否则继续重试
            try {
                if (onException(msgList)) {
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            } catch (Exception ex) {
                log.error("onException error.", ex);
            }

            // 如果有报错的话，那就返回重试状态，稍后还会执行，最大 16 次
            // 不过也可以通过 consumer.setMaxReconsumeTimes 方法设置重新消费次数，
            // 详细情况，可以进入 org.apache.rocketmq.client.consumer.DefaultMQPushConsumer.maxReconsumeTimes 字段的注释信息看看情况
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
    }
}